package com.yanxu;

import com.yanxu.domain.Event;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.Calendar;
import java.util.Random;

/**
 * @author 折戟沉沙铁未销
 * @version V1.0
 * @date 2025-07-13-2025
 * @Description: api 学习
 */
public class Api04_Source_Custom {
    public static void main(String[] args) throws Exception {
        //获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //源 -算子

        //从自定义 source 中接收数据
        DataStreamSource<Event> streamSource = env.addSource(new CustomSource());
        streamSource.print("custom source >>>");


        //对环境进行执行。
        env.execute();
    }
}


// 注意：使用 SourceFunction 。并行度只能是1 。
//      所以，当数据量比较大的时候，并不是很好
@Slf4j
class CustomSource implements SourceFunction<Event> {

    // 声明一个布尔型变量，用于控制数据源是否继续运行
    private Boolean running = true;


    /**
     * 数据源的核心方法，用于产生数据
     *
     * @param ctx 上下文对象，用于收集产生的数据
     * @throws Exception 可能抛出的异常
     */
    @Override
    public void run(SourceContext<Event> ctx) throws Exception {
        // 创建Random实例，用于生成随机数
        Random random = new Random();
        // 定义用户数组和访问URL数组
        String[] users = {"Mary", "Alice", "Bob", "Cary"};
        String[] urls = {"./home", "./cart", "./fav", "./prod?id=1", "./prod?id=2"};

        // 当running为true时，持续生成事件
        while (running) {
            // 随机选择一个用户和URL，结合当前时间戳，创建一个新的Event对象并收集
            ctx.collect(new Event(
                    users[random.nextInt(users.length)],
                    urls[random.nextInt(urls.length)],
                    Calendar.getInstance().getTimeInMillis()
            ));
            log.error("custom source .....");
            // 为了便于观察和处理，每生成一个事件后暂停1秒
            Thread.sleep(1000);
        }
    }

    /**
     * 取消数据生成的方法，设置running为false停止数据源
     */
    @Override
    public void cancel() {
        running = false;
    }
}